package com.coins.common.jobs;

import java.math.BigDecimal;
import java.util.Map;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.PriorityBlockingQueue;
import java.util.concurrent.TimeUnit;

import org.apache.log4j.Logger;
import org.springframework.beans.factory.DisposableBean;
import org.springframework.beans.factory.InitializingBean;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Component;

import com.alibaba.fastjson.JSONObject;
import com.coins.agent.HTTPEvents;
import com.coins.agent.HTTPManagers;
import com.coins.common.redis.RedisManagers;
import com.coins.common.utils.AntiDuplicateLinkedBlockingQueue;

/**
 * 功能：代理HTTP管理
 */
@Component
public class JobPools implements InitializingBean, DisposableBean {

	protected static final Logger LOGGER = Logger.getLogger(JobPools.class);

	private int threadSize = 20;
	@Autowired
	private HTTPManagers httpManagers;
	@Autowired
	private RedisManagers redisManagers;
	private ExecutorService threadPool;
	private Map<String, BigDecimal> qc = new ConcurrentHashMap<String, BigDecimal>();
	private Map<String, Integer> map = new ConcurrentHashMap<String, Integer>();
	private final PriorityBlockingQueue<JobMessage> msgqueue = new PriorityBlockingQueue<JobMessage>();
	private final PriorityBlockingQueue<JobMessage> resqueue = new PriorityBlockingQueue<JobMessage>();
	private final AntiDuplicateLinkedBlockingQueue<String, Integer> signalQueue = new AntiDuplicateLinkedBlockingQueue<String, Integer>();

	public Map<String, BigDecimal> getQc() {
		return qc;
	}

	public void setQc(Map<String, BigDecimal> qc) {
		this.qc = qc;
	}

	public Map<String, Integer> getmap() {
		return map;
	}

	public boolean addSignal(String key, Integer value) {
		if (key != null) {
			return this.signalQueue.put(key, value);
		}
		return false;
	}

	public boolean addJobMessage(JobMessage msg) {
		if (msg != null) {
			this.msgqueue.put(msg);
			return true;
		}
		return false;
	}

	@Override
	public void afterPropertiesSet() throws Exception {
		this.threadPool = Executors.newFixedThreadPool(threadSize);
		this.threadPool.execute(new Runnable() {
			@Override
			public void run() {
				while (true) {
					JobMessage msg = null;
					try {
						msg = msgqueue.take();
						Long ptime = Long.valueOf(0);
						if (System.getProperty("POOLSTIME") != null) {
							ptime = Long.valueOf(System.getProperty("POOLSTIME"));
						}
						if (msg.getEvents() == HTTPEvents.ZBTICKER) {
							if (ptime < 200) {
								Thread.sleep(200 - ptime);
							}
						} else if (msg.getEvents() == HTTPEvents.ZBGETORDERSIGNORETRADETYPE) {
							Thread.sleep(10 * 1000);
						} else {
							Thread.sleep(1 * 1000);
						}
						long ben = System.currentTimeMillis();
						httpManagers.onEvents(msg.getEvents(), msg);
						resqueue.put(msg);
						long end = System.currentTimeMillis();
						long time = end - ben;
						if (map.containsKey(msg.getEvents().toString())) {
							Integer count = map.get(msg.getEvents().toString());
							count++;
							map.put(msg.getEvents().toString(), Integer.valueOf(count));
						} else {
							map.put(msg.getEvents().toString(), Integer.valueOf(1));
						}
						System.setProperty("POOLSTIME", String.valueOf(time));
					} catch (Exception e) {
						resqueue.put(msg);
						if (LOGGER.isErrorEnabled())
							LOGGER.error(String.format("请求数据[%s]", e.getMessage()));
					}
				}
			}
		});
		this.threadPool.execute(new Runnable() {
			@Override
			public void run() {
				while (true) {
					JobMessage msg = null;
					try {
						msg = resqueue.take();
						redisManagers.pubMessage(msg);
					} catch (Exception e) {
						LOGGER.error(String.format("处理返回[%s][%s][%s]", msg.getEvents(), msg.getData(), msg.getResult()));
					}
				}
			}
		});
	}

	@Override
	public void destroy() throws Exception {
		if (this.threadPool != null) {
			this.threadPool.shutdown();
			try {
				if (!threadPool.awaitTermination(60, TimeUnit.SECONDS)) {
					threadPool.shutdownNow();
					System.out.println(JSONObject.toJSONString(map));
					if (!threadPool.awaitTermination(60, TimeUnit.SECONDS))
						LOGGER.error("threadPool success destroy");
				}
			} catch (InterruptedException ie) {
				threadPool.shutdownNow();
				Thread.currentThread().interrupt();
			}
		}
	}
}
